package com.ctbri.manage.bydeequ.calculate
//import com.amazon.deequ.VerificationSuite
//import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}

import java.lang.reflect
import com.ctbri.manage.quality.scala.original.{CheckBasic, ConfigReader}
import org.eclipse.jetty.websocket.common.frames.DataFrame
//import org.apache.spark.sql.SparkSession
//import org.apache.spark.sql.types._
//import org.apache.spark.sql._
//import com.amazon.deequ.repository.{MetricsRepository, ResultKey}
import com.google.common.io.Files
//import com.amazon.deequ.repository.fs.FileSystemMetricsRepository
//import com.amazon.deequ.constraints.{ConstrainableDataTypes, ConstraintStatus}
//import org.apache.spark.sql.expressions.Window
//import org.apache.spark.sql.types.{StringType, StructField, StructType}
//import org.apache.spark.sql.functions._

import java.util
import java.io.File
import scala.collection.JavaConverters._
import java.lang.StringBuilder
import scala.collection.mutable
import scala.util.matching.Regex

/**
 * @author songyunlong
 * @createTime 2023/6/5 19:10
 * @description
 */
class CalculateV1(appNamec: String = null,
                  masterSetc: String = null,
                  logLevelc: String = null,
                  //from csv
                  dataExportPathc: String = null,
                  dataSavePathc: String = null,
                  dataSaveStylec: String = null,
                  //from jdbc
                  urlc: String = null,
                  driverc: String = null,
                  userc: String = null,
                  passwordc: String = null,
                  dbtablec: String = null,
                  //from hive
                  databaseNamec: String = null,
                  tableNamec: String = null,
                  //
                  configPathc: String=null,
                  dataSourcec: String=null) extends CheckBasic(appNamec,
    masterSetc, logLevelc, dataExportPathc, dataSavePathc, dataSaveStylec,
    urlc, driverc, userc, passwordc, dbtablec, databaseNamec, tableNamec) with DataListGeneratedBySql{
    var configPath = configPathc
    var dataSource = dataSourcec

    /**
     * excel,csv,json,txt
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param dataExportPath
     * @param dataSavePath
     * @param dataSaveStyle
     * @param configPath
     * @param dataSource
     */
    def this(appName: String,
        masterSet: String,
        logLevel: String,
        dataExportPath: String,
        dataSavePath: String,
        dataSaveStyle: String,
             configPath: String,
             dataSource: String){
        this(appNamec=appName,
            masterSetc=masterSet,
            logLevelc=logLevel,
            dataExportPathc=dataExportPath,
            dataSavePathc=dataSavePath,
            dataSaveStylec=dataSaveStyle,
            configPathc=configPath,
            dataSourcec=dataSource
        )
    }

    /**
     * hive
     *
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param databaseName
     * @param tableName
     * @param dataSavePath
     * @param dataSaveStyle
     * @param configPath
     * @param dataSource
     */
    def this(appName: String,
             masterSet: String,
             logLevel: String,
             databaseName: String,
             tableName: String,
             dataSavePath: String,
             dataSaveStyle: String,
             configPath: String,
             dataSource: String){
        this(appNamec = appName,
            masterSetc = masterSet,
            logLevelc = logLevel,
            databaseNamec = databaseName,
            tableNamec = tableName,
            dataSavePathc = dataSavePath,
            dataSaveStylec = dataSaveStyle,
            configPathc=configPath,
            dataSourcec=dataSource)
    }

    /**
     * jdbc
     *
     * @param appName
     * @param masterSet
     * @param logLevel
     * @param url
     * @param driver
     * @param user
     * @param password
     * @param dbtable
     * @param dataSavePath
     * @param dataSaveStyle
     * @param configPath
     * @param dataSource
     */
    def this(appName: String,
             masterSet: String,
             logLevel: String,
             url: String,
             driver: String,
             user: String,
             password: String,
             dbtable: String,
             dataSavePath: String,
             dataSaveStyle: String,
             configPath: String,
             dataSource: String){
        this(appNamec = appName,
            masterSetc = masterSet,
            logLevelc = logLevel,
            urlc = url,
            driverc = driver,
            userc = user,
            passwordc = password,
            dbtablec = dbtable,
            dataSavePathc = dataSavePath,
            dataSaveStylec = dataSaveStyle,
            configPathc=configPath,
            dataSourcec=dataSource)
    }
    val getSpark = this.dataSource match {
        case "hive" => this.envSetHive(appName=this.appName, masterSet=this.masterSet, logLevel=this.logLevel)
        case _ => this.envSet(appName=this.appName, masterSet=this.masterSet, logLevel=this.logLevel)
    }

  /*  def getDataset(structType: StructType = null): Dataset[Row]={
        if (this.dataSource.equals("excel")){
            if (structType == null) this.getDataFromExcel(path=this.dataExportPath, spark=this.getSpark)
            else this.getDataFromExcel(path=this.dataExportPath, spark=this.getSpark, structType=structType)
        }
        else if (this.dataSource.equals("hive")) this.getDataFromHive(spark=this.getSpark, databaseName=this.databaseName, tableName=this.tableName)
        else null
    }*/

//    def withSpark(func: SparkSession => Unit): Unit = {
//        val session = this.getSpark
//        session.sparkContext.setCheckpointDir(System.getProperty("java.io.tmpdir"))
//        try {
//            func(session)
//        } finally {
//            session.stop()
//            System.clearProperty("spark.driver.port")
//        }
//    }
    def metricsCalculateAndRepository(data: DataFrame): Unit={

        val checkFunces = ConfigReader(configPath=this.configPath, dataSource=this.dataSource).getConfigMap

        //遍历配置文件中每一个稽查功能和功能的参数列表，并通过反射作用到check对象上
        for ((checkFuncName, paramDict) <- mapAsScalaMap(checkFunces)){
            println("稽查函数名称: " + checkFuncName)
            println("参数列表: " + paramDict)
            if (checkFuncName.equals("hasSize")){
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion+assertionParam.get("thv"))
                val assertion: Long => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toInt
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toInt
                    case "greaterThan" => x => x > assertionParam.get("thv").toInt
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toInt
                    case "equalsTo" => x => x == assertionParam.get("thv").toInt
                }
              /*  val check = Check(CheckLevel.Warning, checkFuncName)
                    .hasSize(assertion=assertion)
                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)*/
            }
            if (checkFuncName.equals("hasCompleteness")){
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasCompleteness(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticNull(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath+checkFuncName+"/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("isUnique")){ //针对某一列，所有空值行不计入的情况下，统计有多少行的值是唯一的，输出列中值唯一的行数与去除空值后的总行数的比值
                val column = paramDict.get("column").asInstanceOf[String]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .isUnique(column)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticUniqueofOneColumn(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath+checkFuncName+"/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasMin")){ //需要转换为数值类型后再用此功能, 去除空值行之后得到最小值，目前只能用于数值型的列
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasMin(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticMin(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasMax")) { //需要转换为数值类型后再用此功能, 去除空值行之后得到最大值，目前只能用于数值型的列
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasMax(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticMax(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasMean")) { //需要转换为数值类型后再用此功能, 去除空值行之后得到平均值，目前只能用于数值型的列
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasMean(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticMean(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasSum")) { //需要转换为数值类型后再用此功能, 去除空值行之后得到最小值，目前只能用于数值型的列
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasSum(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticSum(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("containsEmail")){ // 在不去除空值行的情况下判断每行是否与邮件字符串形式匹配，返回匹配的行数与总行数的比值
                val column = paramDict.get("column").asInstanceOf[String]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .containsEmail(column)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticContainsEmail(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("containsURL")) { // 在不去除空值行的情况下判断每行是否与URL形式匹配，返回匹配的行数与总行数的比值
                val column = paramDict.get("column").asInstanceOf[String]
              /*  val check = Check(CheckLevel.Warning, checkFuncName)
                    .containsURL(column)
                this.resultGeneration(check, data, checkFuncName, "x=1.0")*/
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticContainsURL(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("containsCreditCardNumber")) { // 在不去除空值行的情况下判断每行是否与身份证件形式匹配，返回匹配的行数与总行数的比值
                val column = paramDict.get("column").asInstanceOf[String]
              /*  val check = Check(CheckLevel.Warning, checkFuncName)
                    .containsCreditCardNumber(column)
                this.resultGeneration(check, data, checkFuncName, "x=1.0")*/
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticContainsCreditCardNumber(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
//            if (checkFuncName.equals("hasDataType")){ //检查以针对符合给定数据类型的行的部分运行
//                val column = paramDict.get("column").asInstanceOf[String]
//                val dataType: ConstrainableDataTypes.Value = paramDict.get("dataType").asInstanceOf[String] match {
//                    case "Numeric" => ConstrainableDataTypes.Numeric
//                    case "String" => ConstrainableDataTypes.String
//                    case "Boolean" => ConstrainableDataTypes.Boolean
//                    case "Integral" => ConstrainableDataTypes.Integral
//                    case "Fractional" => ConstrainableDataTypes.Fractional
//                    case "Null" => ConstrainableDataTypes.Null
//                    case _ => null
//                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasDataType(column, dataType)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
//            }
            if (checkFuncName.equals("isContainedIn")){ //需要列值是String类型的, 判断各行为空值或者取值包含在给定列表中的数与总行数的比值
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
                val allowedValues = paramDict.get("allowedValues").asInstanceOf[util.ArrayList[String]]
               /* val check = Check(CheckLevel.Warning, checkFuncName)
                    .isContainedIn(column, asScalaBuffer(allowedValues).toArray[String], assertion)
                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)*/
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticIsContainedIn[String](meetRequirements, this.getSpark, column, asScalaBuffer(allowedValues).toArray[String], data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasPattern")){ // 非空且满足正则匹配条件的行数与总行数的比值
                val column = paramDict.get("column").asInstanceOf[String]
                val pattern = paramDict.get("pattern").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasPattern(column, new Regex(pattern))
               // this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticHasPattern(meetRequirements, this.getSpark, column, pattern, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("isLessThanOrEqualTo")){ // 判断列A是否小于等于列B, 这其中需要两列中任意一列存在空值的行都记为不满足要求，计算满足条件的函数与总行数的比值
                val columnA = paramDict.get("columnA").asInstanceOf[String]
                val columnB = paramDict.get("columnB").asInstanceOf[String]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .isLessThanOrEqualTo(columnA, columnB)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticIsLessThanOrEqualTo(meetRequirements, this.getSpark, columnA, columnB, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("isLessThan")){ // 判断列A是否小于列B, 这其中需要两列中任意一列存在空值的行都记为不满足要求，计算满足条件的函数与总行数的比值
                val columnA = paramDict.get("columnA").asInstanceOf[String]
                val columnB = paramDict.get("columnB").asInstanceOf[String]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .isLessThan(columnA, columnB)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticIsLessThan(meetRequirements, this.getSpark, columnA, columnB, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("isGreaterThan")) { // 判断列A是否大于列B, 这其中需要两列中任意一列存在空值的行都记为不满足要求，计算满足条件的函数与总行数的比值
                val columnA = paramDict.get("columnA").asInstanceOf[String]
                val columnB = paramDict.get("columnB").asInstanceOf[String]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .isGreaterThan(columnA, columnB)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticIsGreaterThan(meetRequirements, this.getSpark, columnA, columnB, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("isGreaterThanOrEqualTo")) { // 判断列A是否大于等于列B, 这其中需要两列中任意一列存在空值的行都记为不满足要求，计算满足条件的函数与总行数的比值
                val columnA = paramDict.get("columnA").asInstanceOf[String]
                val columnB = paramDict.get("columnB").asInstanceOf[String]
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .isGreaterThanOrEqualTo(columnA, columnB)
//                this.resultGeneration(check, data, checkFuncName, "x=1.0")
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
               // val dataList = this.staticIsGreaterThanOrEqualTo(meetRequirements, this.getSpark, columnA, columnB, data)
               // this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }

            if (checkFuncName.equals("hasMinLength")){ //在空值行忽略不计的条件下，判断最小长度字符串是否满足assertion不等式关系
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasMinLength(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticHasMinLength(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasMaxLength")){ //在空值行忽略不计的条件下，判断最大长度字符串是否满足assertion不等式关系
                val column = paramDict.get("column").asInstanceOf[String]
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasMaxLength(column, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val dataList = this.staticHasMaxLength(meetRequirements, this.getSpark, column, data)
//                this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasColumnsCount")){
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Long => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toInt
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toInt
                    case "greaterThan" => x => x > assertionParam.get("thv").toInt
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toInt
                    case "equalsTo" => x => x == assertionParam.get("thv").toInt
                }
//                val columnNum = data.columns.size
//                val message = assertion(columnNum) match {
//                    case true => "ColumnCountConstraint(Size(None)) successed: Value meet the constraint requirement"
//                    case false => s"ColumnCountConstraint(Size(None)) failed: Value: ${columnNum} does not meet the constraint requirement!"
//                }
//                this.resultGenerationForCustomFunc(columnNum.toString, checkFuncName, assertionExpress.toString, message)
            }
            if (checkFuncName.equals("containsColumns")){ //todo: totalColNum需要改成yaml的集合中的总待检查字段数量
                val columnNames = paramDict.get("columnNames").asInstanceOf[util.ArrayList[String]]
//                val columnsOfDataset = data.columns.toSet
                val messagePrefix = new StringBuilder(")included in the column name of the data table")
                val messagePostfix = new scala.StringBuilder(")not included in the column name of the data table")
                var isSuccessful = true
                var isContainsIn = 0.0
                var isNotContainsIn = 0.0
//                columnNames.forEach {
//                    col => {
//                        if (columnsOfDataset.contains(col)){
//                            messagePrefix.insert(0, col+" ")
//                            isContainsIn += 1
//                        }
//                        else{
//                            messagePostfix.insert(0, col)
//                            isNotContainsIn += 1
//                            if (isSuccessful){
//                                isSuccessful = false
//                            }
//                        }
//                    }
//                }
//                val totalColNum = data.columns.size
                val message = isSuccessful match{
                    case true => s"contaionsColumns successed: 1.0 (${messagePrefix}"
                    case false => s"contaionsColumns failed: ${(isContainsIn/columnNames.size())}(${messagePrefix}, ${isNotContainsIn/columnNames.size()}(${messagePostfix}"
                }
                this.resultGenerationForCustomFunc("*", checkFuncName=checkFuncName, assertionExpress="x=1.0", message=message)
            }
            if (checkFuncName.equals("haveCompleteness")) { // 所有待检测字段均不为空的行数是否满足assertion不等式关系
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
                val columns = asScalaBuffer(paramDict.get("columns").asInstanceOf[util.ArrayList[String]])
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .haveCompleteness(columns, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
               // val dataList = this.staticHaveCompleteness(meetRequirements, this.getSpark, columns.toArray[String], data)
                //this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasUniqueness")) { //针对多列，所有待检查字段同时为空值行不计入（不全为空时需要计入）的情况下，统计有多少行的多列组合值是唯一的，输出唯一的行数与不计入同时为空的总行数的比值
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
                val columns = asScalaBuffer(paramDict.get("columns").asInstanceOf[util.ArrayList[String]])
//                val check = Check(CheckLevel.Warning, checkFuncName)
//                    .hasUniqueness(columns, assertion)
//                this.resultGeneration(check, data, checkFuncName, assertionExpress.toString)
                //根据需求生成数据清单
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
               // val dataList = this.staticHasUniqueness(meetRequirements, this.getSpark, columns.toArray[String], data)
              //  this.resultSave(dataList, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
            if (checkFuncName.equals("hasOneColCompare")) { //去除空值行后，对单字段运用自定义函数统计结果，输出满足条件的行数与不包含空值行的总行数的比值
                //assertion配置解析
                val assertionExpress = new StringBuilder("x")
                val assertionParam = paramDict.get("assertion").asInstanceOf[util.HashMap[String, String]]
                val compartion = assertionParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                assertionExpress.append(compartion + assertionParam.get("thv"))
                val assertion: Double => Boolean = assertionParam.get("compareStyle") match {
                    case "lessThan" => x => x < assertionParam.get("thv").toDouble
                    case "lessOrEqualThan" => x => x <= assertionParam.get("thv").toDouble
                    case "greaterThan" => x => x > assertionParam.get("thv").toDouble
                    case "greaterOrEqualThan" => x => x >= assertionParam.get("thv").toDouble
                    case "equalsTo" => x => x == assertionParam.get("thv").toDouble
                }
                //单字段值比较函数配置解析
                val compareFuncExpress = new StringBuilder("x")
                val compareFuncParam = paramDict.get("compareFunc").asInstanceOf[util.HashMap[String, String]]
                val compareFuncCompartion = compareFuncParam.get("compareStyle") match {
                    case "lessThan" => "<"
                    case "lessOrEqualThan" => "<="
                    case "greaterThan" => ">"
                    case "greaterOrEqualThan" => ">="
                    case "equalsTo" => "="
                }
                compareFuncExpress.append(compareFuncCompartion + compareFuncParam.get("thv"))
                //
                val column = paramDict.get("column").asInstanceOf[String]
                val meetRequirements = paramDict.get("meetRequirements").asInstanceOf[Boolean]
//                val res = this.staticHasOneColCompare(meetRequirements, this.getSpark, column, compareFuncParam, data)
//                val result = res._3 / res._2
//                val message = assertion(result) match {
//                    case true => "HasOneColCompare(Size(None)) successed: Value meet the constraint requirement"
//                    case false => s"HasOneColCompare(Size(None)) failed: Value: ${result} does not meet the constraint requirement!"
//                }
//                this.resultGenerationForCustomFunc(result.toString, checkFuncName, assertionExpress.toString, message, "Column", "Comparison")
               // this.resultSave(res._1, this.dataSavePath + checkFuncName + "/dataList", this.dataSaveStyle)
            }
        }
       // this.getSpark.stop()
    }
    def resultGenerationForCustomFunc(result: String, checkFuncName: String, assertionExpress: String, message: String): Unit ={
//        val rdd = this.getSpark.sparkContext.parallelize(Seq(ConfigInfoTotal(
//            "Dataset",
//            "*",
//            result,
//            assertionExpress,
//            message
//        )))
//        val data = this.getSpark.createDataFrame(rdd)
//        data.show(100, false)
//        data.repartition(1).write
//            .mode("overwrite")
//            .format("csv")
//            .option("header", "true")
//            .save(this.dataSavePath + checkFuncName)
    }

    def resultGenerationForCustomFunc(result: String, checkFuncName: String, assertionExpress: String, message: String, entity: String, instance: String): Unit = {
//        val rdd = this.getSpark.sparkContext.parallelize(Seq(ConfigInfoTotal(
//            entity,
//            instance,
//            result,
//            assertionExpress,
//            message
//        )))
//        val data = this.getSpark.createDataFrame(rdd)
//        data.show(100, false)
//        data.repartition(1).write
//            .mode("overwrite")
//            .format("csv")
//            .option("header", "true")
//            .save(this.dataSavePath + checkFuncName)
    }
   /* def resultGeneration(check: Check, data: Dataset[Row], checkFuncName: String, assertionExpress: String=null): Unit={
        //
        val metricsFile = new File(Files.createTempDir(), "metrics.json")
        val repository: MetricsRepository = FileSystemMetricsRepository(this.getSpark, metricsFile.getAbsolutePath)
        val resultKey = ResultKey(System.currentTimeMillis(), Map("tag" -> "repository"))
        val verificationResult = VerificationSuite().onData(data = data)
            .addCheck(check)
            .useRepository(repository)
            .saveOrAppendResult(resultKey)
            .run()
        val resultsForAllConstraints = verificationResult.checkResults
            .flatMap { case (_, checkResult) => checkResult.constraintResults }

        val resultsForAllConstraintsForDF = resultsForAllConstraints
            .map { result =>
                result.status match {
                    case ConstraintStatus.Success => s"${result.constraint} successed: Value meet the constraint requirement"
                    case ConstraintStatus.Failure => s"${result.constraint} failed: ${result.message.get}"
                }
            }
//        println(assertionExpress)

        val rdd2 = this.getSpark.sparkContext.parallelize(Seq(ConfigInfo(assertionExpress, resultsForAllConstraintsForDF.head)))
        val data2 = this.getSpark.createDataFrame(rdd2)
            .select(
            row_number().over(Window.orderBy(lit(1))).alias("b_id"), col("assertion"), col("message"))
//        data2.show(100, false)
        val seccessMetric = repository.load()
            .withTagValues(Map("tag" -> "repository"))
            .getSuccessMetricsAsDataFrame(this.getSpark)
            .drop("tag")
            .withColumn(colName = "a_id", row_number().over(Window.orderBy(lit(1))))
//        seccessMetric.show(100, false)
        val finalResult = seccessMetric.join(right = data2,
            joinExprs = seccessMetric.col("a_id") === data2.col("b_id"),
            joinType = "left")
            .drop("name").drop("a_id").drop("b_id").drop("dataset_date")
        finalResult.show(100, false)
        finalResult.repartition(1).write
            .mode("overwrite")
            .format("csv")
            .option("header", "true")
            .save(this.dataSavePath + checkFuncName)
    }*/
}

object CalculateV1{
    def main(args: Array[String]): Unit={
        val javaMap = ConfigReader(configPath = "/Users/songyunlong/Desktop/data-manage/data-quality/src/main/resources/config/sub/deequConfig.yaml",
            dataSource = "excel").getConfigMap()
        val isContainedIn1 = javaMap.get("isContainedIn")
        println(isContainedIn1)
//
    }
}

/**
 *
 * @param assertion: String
 * @param message: String
 */
case class ConfigInfo(assertion: String, message: String)

/**
 *
 * @param entity: String
 * @param instance: String
 * @param value: String
 * @param assertion: String
 * @param message: String
 */
case class ConfigInfoTotal(entity: String, instance: String, value: String, assertion: String, message: String)
